package practice.phase04.module02;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * @author Alex Zhang
 */
public class KafkaTest {
    private static Properties consumerConfig;
    @BeforeAll
    private static void init() {
        consumerConfig = new Properties();
        consumerConfig.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "node1:9092");
        consumerConfig.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumerConfig.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        consumerConfig.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "grp");
        consumerConfig.setProperty(ConsumerConfig.CLIENT_ID_CONFIG, "clentid");
        consumerConfig.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    }
    @Test
    public void interceptiorTest() {
        final String interceptors = "phase04.module02.code.consumerinterceptors.ConsumerInterceptorExample1"
                + ",phase04.module02.code.consumerinterceptors.ConsumerInterceptorExample2"
                + ",phase04.module02.code.consumerinterceptors.ConsumerInterceptorExample3";
        consumerConfig.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerConfig);
        consumer.subscribe(Collections.singleton("tp_demo"));
        int count = Integer.MAX_VALUE;
        while (count >= 0) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));
            records.forEach(record -> {
                System.out.println(record.topic());
                System.out.println(record.partition());
                System.out.println(record.offset());
            });
            count -= 1;
        }
        consumerConfig.remove(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG);
    }
}
